package com.codejiwei.flink.broadcast;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
import org.apache.http.HttpEntity;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.util.EntityUtils;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * author: codejiwei
 * date: 2023/8/2
 * desc:
 **/
public class RuleBroadcastFetcher extends RichParallelSourceFunction<Map<String, String>> {
    private volatile boolean isRunning;

    @Override
    public void open(Configuration parameters) throws Exception {
        isRunning = true;
    }

    @Override
    public void run(SourceContext<Map<String, String>> sourceContext) throws Exception {

        HashMap<String, String> rules = new HashMap<>();

        while (isRunning) {

            rules.clear();

            CloseableHttpClient httpClient = HttpClients.createDefault();
            HttpGet httpGet = new HttpGet("http://localhost:8111/api/distributor/userrulestips");
            RequestConfig requestConfig = RequestConfig.custom().setConnectTimeout(5000).setSocketTimeout(600000).build();
            httpGet.setConfig(requestConfig);
            CloseableHttpResponse response = httpClient.execute(httpGet);
            HttpEntity entity = response.getEntity();
            String result = EntityUtils.toString(entity);

            Map<String, Object> resp = JsonUtil.decode(result);
            int errno = Integer.valueOf(resp.get("errno").toString());

            if (errno == 0) {
                Map<String, List<String>> batchData = (Map<String, List<String>>) resp.get("data");
                for (Map.Entry<String, List<String>> entry : batchData.entrySet()) {
                    String rule = entry.getKey();
                    List<String> value = entry.getValue();
                    for (String s : value) {
                        rules.put(rule, s);
                    }
                }

            }

            sourceContext.collect(rules);
            Thread.sleep(2000);

        }

    }

    @Override
    public void cancel() {
        isRunning = false;
    }
}
